package realtime.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import realtime.bean.TableProcess;
import realtime.util.JdbcUtil;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.*;

public class DwdTableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private HashMap<String, TableProcess> openHashMap;

    public DwdTableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        openHashMap = new HashMap<>();
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall_config?" +
                "user=root&password=123456&useUnicode=true&" +
                "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        String sql = "select * from table_process where sink_type = 'dwd'";

        List<TableProcess> tableProcesses = JdbcUtil.queryList(conn, sql, TableProcess.class, true);
        for (TableProcess tableProcess : tableProcesses) {
            openHashMap.put(tableProcess.getSourceTable() + "_" + tableProcess.getSourceType(), tableProcess);
        }

    }
    //Value:{"before":null,"after":{"source_table":"base_category3","sink_table":"dim_base_category3","sink_columns":"id,name,category2_id","sink_pk":"id","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1669162876406,"snapshot":"false","db":"gmall-220623-config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1669162876406,"transaction":null}
    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        JSONObject jsonObject = JSON.parseObject(value);
        if ("d".equals(jsonObject.getString("op"))){
            TableProcess before = JSON.parseObject(jsonObject.getString("before"), TableProcess.class);
            if ("dwd".equals(before.getSinkType())){
                broadcastState.remove(before.getSourceType()+ "_" + before.getSourceType());
                openHashMap.remove(before.getSourceType() + "_" + before.getSourceType());
            }
        }else {
            TableProcess after = JSON.parseObject(jsonObject.getString("after"), TableProcess.class);
            if ("dwd".equals(after.getSinkType())){
                broadcastState.put(after.getSourceTable() + "_" + after.getSourceType(),after);
            }
        }
    }
    //Value:{"database":"gmall-220623-flink","table":"base_trademark","type":"insert","ts":1669162958,"xid":1111,"xoffset":13941,"data":{"id":1595211185799847960,"user_id":119,"nick_name":null,"head_img":null,"sku_id":31,"spu_id":10,"order_id":987,"appraise":"1204","comment_txt":"评论内容：48384811984748167197482849234338563286217912223261","create_time":"2022-08-02 08:22:38","operate_time":null}}

    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        String key = value.getString("table") + "_" + value.getString("type");
        TableProcess tableProcess = broadcastState.get(key);
        TableProcess mapTableProcess = openHashMap.get(key);

        if (tableProcess != null || mapTableProcess != null){
            if (tableProcess == null){
                tableProcess = mapTableProcess;
            }
            String data = value.getString("data");
            filterColumns(tableProcess,data);

            value.put("sink_table",tableProcess.getSinkTable());
            out.collect(value);
        }else {
            System.out.println("配置类型不匹配：" + key);
        }

    }

    private void filterColumns(TableProcess tableProcess, String data) {
        JSONObject jsonObject = JSON.parseObject(data);
        Set<Map.Entry<String, Object>> entries = jsonObject.entrySet();

        String[] split = tableProcess.getSinkColumns().split(",");
        List<String> list = Arrays.asList(split);

        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
        while (iterator.hasNext()){
            Map.Entry<String, Object> next = iterator.next();
            if (!list.contains(next.getKey())){
                iterator.remove();
            }
        }

    }


}
